package com.shujia.flink.state

import java.util.Properties

import org.apache.flink.api.common.functions.{RichFlatMapFunction, RuntimeContext}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object Demo1ValueState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id", "asdasdddddasd")
    properties.setProperty("auto.offset.reset", "earliest") //earliest 读取所有数据、latest ： 度最新数据


    val consumer = new FlinkKafkaConsumer[String]("test_topic1", new SimpleStringSchema(), properties)

//    consumer.setStartFromLatest()


    val kafkaDS: DataStream[String] = env.addSource(consumer)


    val kvDS: KeyedStream[(String, Int), String] = kafkaDS.map((_, 1)).keyBy(_._1)


    //    val count: DataStream[(String, Int)] = kvDS.sum(1)


    //使用自定义状态统计单词的数量
    val countDS: DataStream[(String, Int)] = kvDS.flatMap(new MyFlatMapFunction)

    countDS.print()


    env.execute()


  }
}

class MyFlatMapFunction extends RichFlatMapFunction[(String, Int), (String, Int)] {


  var valueState: ValueState[Int] = _

  /**
    *
    * 在flatMap 之前执行
    * 在open 方法中定义状态
    *
    */
  override def open(parameters: Configuration): Unit = {

    //获取flink 的执行环境
    val context: RuntimeContext = getRuntimeContext

    //创建状态描述对象
    val valueStateDesc = new ValueStateDescriptor[Int]("count", classOf[Int])


    /**
      * ValueState : 为每一个key 保存一个单一的值
      * 用于保存单纯的数量
      *
      */
    valueState = context.getState(valueStateDesc)

  }

  override def flatMap(value: (String, Int), out: Collector[(String, Int)]): Unit = {


    ///更新单词的数量,使用当前的数量加上之前的数量得到新的数量


    //获取状态的保存的值
    val lastCount: Int = valueState.value()


    //更新状态
    valueState.update(lastCount + value._2)

    //获取最新的数量

    val count: Int = valueState.value()


    //将数据发送到下游
    out.collect((value._1, count))

  }
}

